﻿///////////////////////////////////////////////////////////////////////////////
//
//  This source is part of the  "Managed Task Framework.NET Runtime"  (MTF.NET Runtime).
//
//
//  Copyright (c) 2011, Multicore Consulting Limited (UK)
//  All rights reserved.
//  http://www.multicoreconsulting.co.uk
//
//  Redistribution and use in source and binary forms, with or without
//  modification, are permitted provided that the following conditions
//  are met:
//
//  -  Redistributions of source code must retain the above copyright
//     notice, this list of conditions and the following disclaimer.
//  -  Redistributions in binary form must reproduce the above copyright
//     notice, this list of conditions and the following disclaimer in
//     the documentation and/or other materials provided with the
//     distribution.
//  -  Neither the name of Multicore Consulting Limited nor the names of
//     its contributors may be used to endorse or promote products
//     derived from this software without specific prior written
//     permission.
//
//  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
//  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
//  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
//  FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
//  COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
//  INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
//  BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
//  LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
//  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
//  LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
//  ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
//  POSSIBILITY OF SUCH DAMAGE.
//
///////////////////////////////////////////////////////////////////////////////
//#define AGGREGATOR_TRACE

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using MTFOrg.MTF.MTFRuntime;
using MTFOrg.MTF.MTFCore.Messages;
using System.Diagnostics;

namespace MTFOrg.MTF.MTFCore
{
    /// <summary>
    /// Class implementing the aggregator object.
    /// </summary>
    partial class Aggregator
    {
#if AGGREGATOR_TRACE
        const string TRACE_OBJ = "Agg2";
#endif
        /// <summary>
        /// Holds state for a single element of the aggregator.
        /// </summary>
        protected class AggregatorState
        {
            /// <summary>
            /// Queue of queries received at the producer end.
            /// </summary>
            protected Queue<int> m_QueuedQueries;

            /// <summary>
            /// Records if we are currently querying the first link.
            /// </summary>
            protected bool m_QueryingFirstLink;

            /// <summary>
            /// Contains enable flags for each ordered link.
            /// </summary>
            protected bool[] m_IsOrderedLinkDisabled;

            /// <summary>
            /// Contains enable flags for each unordered link.
            /// </summary>
            protected bool[] m_IsUnorderedLinkDisabled;

            /// <summary>
            /// Mapping from ordered to unordered links.
            /// </summary>
            protected int[] m_ToUnordered;

            /// <summary>
            /// Mapping from unordered to ordered links.
            /// </summary>
            protected int[] m_ToOrdered;

            /// <summary>
            /// Gets and sets the queue of queries received at the producer end.
            /// </summary>
            public Queue<int> QueuedQueries
            {
                get { return m_QueuedQueries; }
                set { m_QueuedQueries = value; }
            }

            /// <summary>
            /// Gets or sets whether the first link is being queried.
            /// </summary>
            public bool QueryingFirstLink
            {
                get { return m_QueryingFirstLink; }
                set { m_QueryingFirstLink = value; }
            }

            /// <summary>
            /// Gets or sets ordered link disable flags.
            /// </summary>
            public bool[] IsOrderedLinkDisabled
            {
                get { return m_IsOrderedLinkDisabled; }
                set { m_IsOrderedLinkDisabled = value; }
            }

            /// <summary>
            /// Gets or sets unordered link disable flags.
            /// </summary>
            public bool[] IsUnorderedLinkDisabled
            {
                get { return m_IsUnorderedLinkDisabled; }
                set { m_IsUnorderedLinkDisabled = value; }
            }

            /// <summary>
            /// Gets or sets the mapping to unordered links.
            /// </summary>
            public int[] ToUnordered
            {
                get { return m_ToUnordered; }
                set { m_ToUnordered = value; }
            }

            /// <summary>
            /// Gets or sets the mapping to ordered links.
            /// </summary>
            public int[] ToOrdered
            {
                get { return m_ToOrdered; }
                set { m_ToOrdered = value; }
            }

            /// <summary>
            /// Constructor.
            /// </summary>
            public AggregatorState()
            {
                m_QueuedQueries = new Queue<int>();
                m_QueryingFirstLink = false;
            }
        }

        /// <summary>
        /// Invoked to initialize this task.
        /// </summary>
        public override void OnInitialize()
        {
            base.OnInitialize();

            for (int i = 0; i < m_ProducerConsumerTask.NumElements; i++)
            {
                AggregatorState aggregatorState = new AggregatorState();

                aggregatorState.IsOrderedLinkDisabled = new bool[m_ProducerConsumerTask.NumConsumerLinks];
                aggregatorState.IsUnorderedLinkDisabled = new bool[m_ProducerConsumerTask.NumConsumerLinks];
                aggregatorState.ToOrdered = new int[m_ProducerConsumerTask.NumConsumerLinks];
                aggregatorState.ToUnordered = new int[m_ProducerConsumerTask.NumConsumerLinks];

                // Pre-calculate and store link ordering
                IConnection firstConnection = m_ProducerConsumerTask.GetConsumerConnectionFromLinkNumber(0);
                for (int link = 0; link < m_ProducerConsumerTask.NumConsumerLinks; link++)
                {
                    aggregatorState.ToUnordered[link] = ConnectionHelper.UnorderedLinkFromOrderedLink(firstConnection, link);
                    aggregatorState.ToOrdered[link] = ConnectionHelper.OrderedLinkFromUnorderedLink(firstConnection, link);
                    aggregatorState.IsOrderedLinkDisabled[link] = ConnectionHelper.IsConsumerLinkDisabled(firstConnection, i, aggregatorState.ToUnordered[link]);
                    aggregatorState.IsUnorderedLinkDisabled[link] = ConnectionHelper.IsConsumerLinkDisabled(firstConnection, i, link);
                }

                m_ProducerConsumerTask.SetElementState(i, aggregatorState);
            }

            // Hook up to the producer-consumer event notifications
            m_ProducerConsumerTask.QueryReceived += new ProducerConsumer.OnQuery(OnQueryReceived);
            m_ProducerConsumerTask.ReleaseReceived += new ProducerConsumer.OnRelease(OnReleaseReceived);
            m_ProducerConsumerTask.EventReceived += new ProducerConsumer.OnEvent(OnEventReceived);
            m_ProducerConsumerTask.ReleasedReceived += new ProducerConsumer.OnReleased(OnReleasedReceived);
        }

        /// <summary>
        /// Handles received query.
        /// </summary>
        /// <param name="producerLinkNum">Link that query was received on</param>
        /// <param name="consumerSlots">Consumer slot states (not used)</param>
        /// <param name="producerSlots">Slot states</param>
        /// <param name="elementState">State associated with the current element</param>
        /// <param name="elementNumber">Number of the current element</param>
        public void OnQueryReceived(int producerLinkNum, Slots slots, int consumerSlot, int producerSlot, ref object elementState, int elementNumber)
        {
#if AGGREGATOR_TRACE
			string statii = string.Empty;
			if (consumerSlot != Slots.NO_SLOT)
			{
				for (int _i = 0; _i < m_ProducerConsumerTask.NumConsumerLinks; _i++)
					statii += " " + (
                        slots.GetConsumerLinkStatus(consumerSlot, _i) == LinkStatus.Inactive ? "I" :
						slots.GetConsumerLinkStatus(consumerSlot, _i) == LinkStatus.Opened ? "O" :
						slots.GetConsumerLinkStatus(consumerSlot, _i) == LinkStatus.Waiting ? "W" : "R");
			}
			if (TRACE_OBJ == string.Empty || TRACE_OBJ == Name) Debug.Print(Name + "[{0}]: Query aggregator link" + statii, elementNumber);
#endif
            AggregatorState aggregatorState = elementState as AggregatorState;

            // Store the producer link number in the producer slots tag
            slots.SetProducerTag(producerSlot, producerLinkNum);

            // Find first ordered enabled link
            int firstOrderedLink = 0;
            while (firstOrderedLink < m_ProducerConsumerTask.NumConsumerLinks && aggregatorState.IsOrderedLinkDisabled[firstOrderedLink]) firstOrderedLink++;

            // Attempt to send a query on the first consumer link
            if (firstOrderedLink < m_ProducerConsumerTask.NumConsumerLinks)
            {
                int firstUnorderedLink = aggregatorState.ToUnordered[firstOrderedLink];

                if (aggregatorState.QueryingFirstLink || !m_ProducerConsumerTask.SendQuery(firstUnorderedLink, producerSlot, elementNumber))
                {
                    // If there is no free slot then add the request to the queue
                    aggregatorState.QueuedQueries.Enqueue(producerSlot);
                }
                else
                {
#if AGGREGATOR_TRACE
                    if (TRACE_OBJ == string.Empty || TRACE_OBJ == Name) Debug.Print(Name + "[{0}]: Query sent on aggregator link {1} order {2}", elementNumber, firstUnorderedLink, 0);
#endif
                    // Clear disabled links
                    for (int i = 0; i < m_ProducerConsumerTask.NumConsumerLinks; i++)
                    {
                        if (aggregatorState.IsUnorderedLinkDisabled[i])
                            m_ProducerConsumerTask.ClearLink(i, elementNumber);
                    }

                    aggregatorState.QueryingFirstLink = true;
                }
            }
        }

        /// <summary>
        /// Handles received release.
        /// </summary>
        /// <param name="producerLinkNum">Link that query was received on</param>
        /// <param name="consumerSlots">Consumer slot states (not used)</param>
        /// <param name="producerSlots">Slot states</param>
        /// <param name="eventToRelease">The event to be released</param>
        /// <param name="isRollback">Whether the release was a roll-back</param>
        /// <param name="elementState">State associated with the current element</param>
        /// <param name="elementNumber">Number of the current element</param>
        /// <returns>Whether to complete the current slot set</returns>
        public bool OnReleaseReceived(int producerLinkNum, Slots slots, int consumerSlot, int producerSlot, Event eventToRelease, bool isRollback, bool deletedSlot, ref object elementState, int elementNumber)
        {
#if AGGREGATOR_TRACE
			string statii = string.Empty;
			if (consumerSlot != Slots.NO_SLOT)
			{
				for (int _i = 0; _i < m_ProducerConsumerTask.NumConsumerLinks; _i++)
					statii += " " + (
                        slots.GetConsumerLinkStatus(consumerSlot, _i) == LinkStatus.Inactive ? "I" :
						slots.GetConsumerLinkStatus(consumerSlot, _i) == LinkStatus.Opened ? "O" :
						slots.GetConsumerLinkStatus(consumerSlot, _i) == LinkStatus.Waiting ? "W" : "R");
			}
            if (TRACE_OBJ == string.Empty || TRACE_OBJ == Name) Debug.Print(Name + "[{0}]: Release aggregator link" + statii, elementNumber);
#endif
            AggregatorState aggregatorState = elementState as AggregatorState;
            EventDictionary[] aggregatedEvents = slots.GetConsumerTag(consumerSlot) as EventDictionary[];

            // Send release to all enabled consumers
            int i = 0;
            while (i < m_ProducerConsumerTask.NumConsumerLinks && aggregatorState.IsUnorderedLinkDisabled[i]) i++;
            while (i < m_ProducerConsumerTask.NumConsumerLinks)
            {
#if AGGREGATOR_TRACE
                if (TRACE_OBJ == string.Empty || TRACE_OBJ == Name) Debug.Print(Name + "[{0}]: Release aggregator link {1}", elementNumber, i);
#endif
                m_ProducerConsumerTask.SendRelease(consumerSlot, aggregatedEvents[i], isRollback, i, elementNumber, true);

                // Find next enabled link
                i++;
                while (i < m_ProducerConsumerTask.NumConsumerLinks && aggregatorState.IsUnorderedLinkDisabled[i]) i++;
            }

            return true;
        }

        /// <summary>
        /// Called when the producer notifies it has finished with an event.
        /// </summary>
        /// <param name="consumerLinkNum">Link that query was received on</param>
        /// <param name="consumerSlots">Consumer slot states (not used)</param>
        /// <param name="producerSlots">Slot states</param>
        /// <param name="elementState">State associated with the current element</param>
        /// <param name="elementNumber">Number of the current element</param>
        public void OnReleasedReceived(int consumerLinkNum, ref object elementState, int elementNumber)
        {
#if AGGREGATOR_TRACE
            if (TRACE_OBJ == string.Empty || TRACE_OBJ == Name) Debug.Print(Name + "[{0}]: Released aggregator link {1}", elementNumber, consumerLinkNum);
#endif
            AggregatorState aggregatorState = elementState as AggregatorState;

            // Find first ordered enabled link
            int firstOrderedLink = 0;
            while (firstOrderedLink < m_ProducerConsumerTask.NumConsumerLinks && aggregatorState.IsOrderedLinkDisabled[firstOrderedLink]) firstOrderedLink++;
            int firstUnorderedLink = aggregatorState.ToUnordered[firstOrderedLink];

            if (consumerLinkNum == firstUnorderedLink)
            {
                // Handle any queued queries
                if (aggregatorState.QueuedQueries.Count > 0 && !aggregatorState.QueryingFirstLink)
                {
                    int queuedProducerSlot = aggregatorState.QueuedQueries.Peek();

                    // Attempt to send a query on the first consumer link
                    if (m_ProducerConsumerTask.SendQuery(firstUnorderedLink, queuedProducerSlot, elementNumber))
                    {
                        aggregatorState.QueuedQueries.Dequeue();
#if AGGREGATOR_TRACE
                        if (TRACE_OBJ == string.Empty || TRACE_OBJ == Name) Debug.Print(Name + "[{0}]: Query sent on aggregator link {1} order {2}", elementNumber, firstUnorderedLink, 0);
#endif
                        // Clear disabled links
                        for (int i = 0; i < m_ProducerConsumerTask.NumConsumerLinks; i++)
                        {
                            if (aggregatorState.IsUnorderedLinkDisabled[i])
                                m_ProducerConsumerTask.ClearLink(i, elementNumber);
                        }

                        aggregatorState.QueryingFirstLink = true;
                    }
#if DEBUG
                    else
                    {
                        // If there is no free slot then an error has occurred
                        throw new Exception("Unable to send queued query when slot should have been available.");
                    }
#endif
                }
            }
        }

        /// <summary>
        /// Called when the ProducerConsumer receives an event
        /// </summary>
        /// <param name="consumerLinkNum">Consumer link (absolute) on which the event arrived</param>
        /// <param name="consumerSlots">Consumer slot state information</param>
        /// <param name="producerSlots">Producer slot state information</param>
        /// <param name="eventReceived">Event that was received</param>
        /// <param name="elementState">State associated with the current element</param>
        /// <param name="elementNumber">Number of the current element</param>
        public void OnEventReceived(int consumerLinkNum, Slots slots, int consumerSlot, int producerSlot, Event eventReceived, ref object elementState, int elementNumber)
        {
#if AGGREGATOR_TRACE
			string statii = string.Empty;
			if (consumerSlot != Slots.NO_SLOT)
			{
				for (int _i = 0; _i < m_ProducerConsumerTask.NumConsumerLinks; _i++)
					statii += " " + (
                        slots.GetConsumerLinkStatus(consumerSlot, _i) == LinkStatus.Inactive ? "I" :
						slots.GetConsumerLinkStatus(consumerSlot, _i) == LinkStatus.Opened ? "O" :
						slots.GetConsumerLinkStatus(consumerSlot, _i) == LinkStatus.Waiting ? "W" : "R");
			}
            if (TRACE_OBJ == string.Empty || TRACE_OBJ == Name) Debug.Print(Name + "[{0}]: Event on aggregator link {1}" + statii, elementNumber, consumerLinkNum);
#endif
            // Insert received event into the aggregated event held in the producer slots structure
            EventDictionary[] aggregatedEvents = slots.GetConsumerTag(consumerSlot) as EventDictionary[];
            if (aggregatedEvents == null)
            {
                aggregatedEvents = new EventDictionary[m_ProducerConsumerTask.NumConsumerLinks];
                slots.SetConsumerTag(consumerSlot, aggregatedEvents);
            }

            aggregatedEvents[consumerLinkNum] = eventReceived as EventDictionary;

            // Get ordered link number
            AggregatorState aggregatorState = elementState as AggregatorState;
            int orderedLink = aggregatorState.ToOrdered[consumerLinkNum];

#if DEBUG
            int numEvs = 0;
            for (int i = 0; i < m_ProducerConsumerTask.NumConsumerLinks; i++)
                if (aggregatedEvents[i] != null) numEvs++;
            for (int i = 0; i <= orderedLink; i++)
                if (!aggregatorState.IsOrderedLinkDisabled[i]) numEvs--;
            if (numEvs != 0) throw new Exception("Inconsistent event count");
#endif

            // If this is the event for the first link in the sequence then reset the QueryingFirstLink flag
            int firstOrderedLink = 0;
            while (firstOrderedLink < m_ProducerConsumerTask.NumConsumerLinks && aggregatorState.IsOrderedLinkDisabled[firstOrderedLink]) firstOrderedLink++;
            if (orderedLink == firstOrderedLink)
            {
                (elementState as AggregatorState).QueryingFirstLink = false;
            }

            // Find next (ordered) link to consume
            int nextLink = orderedLink + 1;
            while (nextLink < m_ProducerConsumerTask.NumConsumerLinks &&
                aggregatorState.IsOrderedLinkDisabled[nextLink])
                nextLink++;

            if (nextLink == m_ProducerConsumerTask.NumConsumerLinks)
            {
                //
                // *** Aggregate events under a single Event Dictionary
                // *** and hierarchy of EventArrays.
                EventDictionary cxnEvents = new EventDictionary();
                for (int linkNum = 0; linkNum < m_ProducerConsumerTask.NumConsumerLinks; linkNum++)
                {
                    int orderedLinkNum = aggregatorState.ToUnordered[linkNum];

                    // Ignore empty events from disabled links
                    if (aggregatedEvents[orderedLinkNum] == null) continue;

                    // Add each item from the individual aggregated events into the output collection.
                    foreach (string tagName in aggregatedEvents[orderedLinkNum].Keys)
                    {
                        // If the output collection does not have an item with this tag
                        if (!cxnEvents.ContainsKey(tagName))
                        {
                            // Add the single event to the output collection
                            cxnEvents.SetEvent(tagName, aggregatedEvents[orderedLinkNum][tagName]);
                        }

                        // Else (if the output collection already has an item with this tag)
                        else
                        {
                            Event previousEvent = cxnEvents[tagName];
                            EventArray previousEvents = previousEvent as EventArray;

                            // If there is a single event in the output collection for this tag
                            if (previousEvents == null)
                            {
                                previousEvents = new EventArray();

                                // Convert to an array
                                previousEvents.Add(previousEvent);
                                cxnEvents.SetEvent(tagName, previousEvents);
                            }

                            // Add each event to the output collection
                            Event eventToAdd = aggregatedEvents[orderedLinkNum][tagName];
                            EventArray eventsToAdd = eventToAdd as EventArray;
                            if (eventsToAdd == null)
                            {
                                previousEvents.Add(eventToAdd);
                            }
                            else
                            {
                                foreach (Event singleEvent in eventsToAdd.Events)
                                {
                                    previousEvents.Add(singleEvent);
                                }
                            }
                        }
                    }
                }

                int producerLinkNum = (int)slots.GetProducerTag(producerSlot);
                m_ProducerConsumerTask.SendEvent(producerLinkNum, producerSlot, cxnEvents, elementNumber);
#if AGGREGATOR_TRACE
                if (TRACE_OBJ == string.Empty || TRACE_OBJ == Name) Debug.Print("{0}[{1}] link {2}", Name, elementNumber, 0);
#endif
            }
            else
            {
                // Otherwise, query on the next sequential consumer slot's connection
                int nextUnorderedLink = aggregatorState.ToUnordered[nextLink];
#if DEBUG
                if (slots.GetConsumerLinkStatus(consumerSlot, nextUnorderedLink) != LinkStatus.Inactive)
                    throw new Exception("Unexpected link state in aggregation.");
#endif
                int slotInUse = slots.GetConsumerSlotsInUse(nextUnorderedLink);
                if (slots.GetConsumerLinkStatus(slotInUse, nextUnorderedLink) == LinkStatus.Inactive)
                {
                    m_ProducerConsumerTask.SendQuery(nextUnorderedLink, producerSlot, elementNumber);
#if AGGREGATOR_TRACE
                    if (TRACE_OBJ == string.Empty || TRACE_OBJ == Name) Debug.Print(Name + "[{0}]: Query sent on aggregator link {1} order {2}", elementNumber, nextUnorderedLink, nextLink);
#endif
                }
            }

            int firstUnorderedLink = aggregatorState.ToUnordered[firstOrderedLink];
            if (consumerLinkNum == firstUnorderedLink)
            {
                // If there is a queued requester then pull them out of state and start another sequence
                if (aggregatorState.QueuedQueries.Count > 0 && !aggregatorState.QueryingFirstLink)
                {
                    if (m_ProducerConsumerTask.SendQuery(firstUnorderedLink, aggregatorState.QueuedQueries.Peek(), elementNumber))
                    {
                        aggregatorState.QueuedQueries.Dequeue();
#if AGGREGATOR_TRACE
                        if (TRACE_OBJ == string.Empty || TRACE_OBJ == Name) Debug.Print(Name + "[{0}]: Query sent on aggregator link {1} order {2}", elementNumber, firstUnorderedLink, 0);
#endif
                        // Clear disabled links
                        for (int i = 0; i < m_ProducerConsumerTask.NumConsumerLinks; i++)
                        {
                            if (aggregatorState.IsUnorderedLinkDisabled[i])
                                m_ProducerConsumerTask.ClearLink(i, elementNumber);
                        }

                        aggregatorState.QueryingFirstLink = true;
                    }
                }
            }
        }
    }
}
